package cgl.narada.transport.rtp;

/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or GlobalMMCS,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called GlobalMMCS,
 * nor may Indiana University or Community Grids Lab or GlobalMMCS appear
 * in their name, without prior written permission of ARTI.
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */

/**
 * RTPLinkManager is used to create/delete RTPLink objects.
 * It listens on a specified topic and performs the appropriate actions based on the messages received.
 * @author Ahmet Uyar
 * @version 1.0
 */

import java.net.InetAddress;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import cgl.narada.jms.NBJmsInitializer;
import cgl.narada.node.BrokerProperties;
import cgl.narada.transport.Link;


// import java.util.logging.Logger;
// import java.util.logging.Level;

public class RTPLinkManager implements MessageListener, RTPLinkChannelStrings {

	// create a logger for this class
	// protected static Logger logger =
	// Logger.getLogger(RTPLinkManager2.class.getName());

	// jms broker port number
	private static int brokerNiotcpPort = 3045;

	// the ip address of the machine this broker is running on
	private InetAddress localIP;

	// topic name to exchange messages
	public static final String baseTopicName = "GlobalMMCS/RtpLinkManager";

	private String linkManagerTopicName;

	// handle for RTPLinkFactory
	private RTPLinkFactory linkFactory;

	// variables to exchange JMS messages
	private TopicSession publisherSession;

	private TopicSession subscriberSession;

	private TopicPublisher basePublisher;

	private TopicSubscriber baseSubscriber;

	private TopicSubscriber linkManagerSubscriber;

	private TopicConnection connection;
	private String moduleName = "RTPLinkManager: ";
	
	public RTPLinkManager(RTPLinkFactory linkFactory) throws Exception {

		this.linkFactory = linkFactory;
		this.localIP = InetAddress.getLocalHost();

		initializeConnection();
	}

	public void initializeConnection() throws Exception {
		BrokerProperties brokerProperties = BrokerProperties.getInstance();
		String portString = brokerProperties.getProperty("NIOTCPBrokerPort");
		if (portString != null) {
			brokerNiotcpPort = Integer.parseInt(portString);
		} else {
			System.out.println(moduleName + "The [NIOTCPBrokerPort] property has"
					+ "not been set. This WILL cause problems.");
		}
		
		NBJmsInitializer ini = new NBJmsInitializer(InetAddress.getLocalHost()
				.getHostAddress(), brokerNiotcpPort);

		// Lookup a JMS connection factory
		TopicConnectionFactory conFactory = (TopicConnectionFactory) ini.lookup();

		// Create a JMS connection
		connection = conFactory.createTopicConnection("guest", "password");

		// Create a JMS session object
		publisherSession = connection.createTopicSession(false,
				Session.AUTO_ACKNOWLEDGE);
		subscriberSession = connection.createTopicSession(false,
				Session.AUTO_ACKNOWLEDGE);

		// create subscriber and publisher for baseTopicName
		Topic baseTopic = publisherSession.createTopic(baseTopicName);
		basePublisher = publisherSession.createPublisher(baseTopic);
		baseSubscriber = subscriberSession.createSubscriber(baseTopic);
		baseSubscriber.setMessageListener(this);

		// create subscriber only for this rtp link
		linkManagerTopicName = buildTopic(this.localIP.getHostAddress());
		Topic linkManagerTopic = subscriberSession
				.createTopic(linkManagerTopicName);
		linkManagerSubscriber = subscriberSession
				.createSubscriber(linkManagerTopic);
		linkManagerSubscriber.setMessageListener(this);

		// Start the JMS connection; allows messages to be delivered
		connection.start();
	}

	/**
	 * set the broker niotcp port number to connect
	 */
	public static void setBrokerNiotcpPort(int niotcpPort) {
		brokerNiotcpPort = niotcpPort;
	}

	/**
	 * build a topic name with the given string it puts a slash to base topic
	 * name and the adds the given string
	 * 
	 * @param extension
	 * @return
	 */
	private String buildTopic(String extension) {
		return baseTopicName + "/" + extension;
	}

	public void onMessage(Message message) {
		try {

			// see whether this is a clear message
			String action = message.getStringProperty(ACTION);
			if (action.equals(CLEAR_RTP_LINK_MANAGER)) {
				logRequestMessage(message, baseTopicName);
				processClear(message);
				return;
			}

			logRequestMessage(message, linkManagerTopicName);
			if (action.equals(START_RTP_LINK)) {
				processStartRTPLink(message);
			} else if (action.equals(START_MULTICAST_RTP_LINK)) {
				processStartMulticastRTPLink(message);
			} else if (action.equals(DELETE_RTP_LINK)) {
				processDeleteRTPLink(message);
			} else if (action.equals(SUBSCRIBE)) {
				processSubscribe(message);
			} else if (action.equals(UNSUBSCRIBE)) {
				processUnsubscribe(message);
			} else if (action.equals(RESPONSE)) {
				System.out.println(action + " received. Do nothing");
			}

		} catch (JMSException jmse) {
			jmse.printStackTrace();
		}

	}

	/**
	 * processes a Clear message
	 * 
	 * Received message will have the following parameters Action:
	 * ClearRTPLinkManager
	 * 
	 * No response message is sent for this.
	 */
	private void processClear(Message message) {

		try {
			// close all links
			linkFactory.closeAllLinks();

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * processes Start message
	 * 
	 * Received message will have the following parameters Action: StartRTPLink
	 * Ip: The ip address of the remote party LocalPort: the port number on which
	 * this receiver will listen to RemotePort: the port number to which this
	 * link will send messages to PublicationTopic: an integer topic number to
	 * which this link publish messages from the remote party.
	 * 
	 * Response message will have the parameters Action: Response
	 * RequestedAction: StartRTPLink Status: Ok or Fail Reason: a string
	 * explaining the problem if it failed. LinkID: linkID of this newly
	 * initialized RTPLink
	 */
	private void processStartRTPLink(Message message) {

		try {
			// retrieve the properties from this message
			String ipStr = message.getStringProperty(IP);
			InetAddress ip = InetAddress.getByName(ipStr);
			int localPort = message.getIntProperty(LOCAL_PORT);
			int remotePort = message.getIntProperty(REMOTE_PORT);
			int publicationTopic = message.getIntProperty(PUBLICATION_TOPIC);

			// create the link
			Link link = linkFactory.createRTPLink(ip, remotePort, localPort,
					publicationTopic);

			if (link == null) {
				sendFailResponse(message, "Cannot create the RTPLink");
				return;
			}

			// send a success response
			sendSuccessResponse(message, link.getLinkId());

		} catch (Exception e) {
			sendFailResponse(message, e.getMessage());
			// e.printStackTrace();
		}
	}

	/**
	 * processes Start message
	 * 
	 * Received message will have the following parameters Action: StartRTPLink
	 * Ip: The ip address of the remote party Port: the port number on which this
	 * receiver will listen to PublicationTopic: an integer topic number to which
	 * this link publish messages from the remote party. TopicRange: the number
	 * of topics this multicsat rtp link can use to publish received streams
	 * 
	 * Response message will have the parameters Action: Response
	 * RequestedAction: StartMulticastRTPLink Status: Ok or Fail Reason: a string
	 * explaining the problem if it failed. LinkID: linkID of this newly
	 * initialized RTPLink
	 */
	private void processStartMulticastRTPLink(Message message) {

		try {
			String ipStr = message.getStringProperty(IP);
			InetAddress ip = InetAddress.getByName(ipStr);
			int port = message.getIntProperty(PORT);
			int publicationTopic = message.getIntProperty(PUBLICATION_TOPIC);
			int topicRange = message.getIntProperty(TOPIC_RANGE);

			Link link = linkFactory.createMulticastRTPLink(ip, port,
					publicationTopic, topicRange);

			if (link == null) {
				sendFailResponse(message, "Cannot create the RTPLink");
				return;
			}

			// send a success response
			sendSuccessResponse(message, link.getLinkId());

		} catch (Exception e) {
			sendFailResponse(message, e.getMessage());
			// e.printStackTrace();
		}
	}

	/**
	 * processes Delete message
	 * 
	 * Received message will have the following parameters Action: DeleteRTPLink
	 * LinkID: linkID of the RTPLink to be deleted
	 * 
	 * Response message will have the parameters Action: Response
	 * RequestedAction: DeleteRTPLink Status: Ok or Fail Reason: a string
	 * explaining the problem if it failed.
	 */
	private void processDeleteRTPLink(Message message) {
		try {
			String linkID = message.getStringProperty(LINKID);
			if (linkID == null) {
				sendFailResponse(message, "LinkID is null.");
				return;
			}

			boolean deleted = linkFactory.closeLink(linkID);
			if (!deleted) {
				sendFailResponse(message, "There is no such link in RTPLinkFactory");
				return;
			}

			// send a success response
			sendSuccessResponse(message);

		} catch (JMSException jmse) {
			// jmse.printStackTrace();
			sendFailResponse(message, jmse.getMessage());
		}
	}

	/**
	 * processes Subscribe message
	 * 
	 * Received message will have the following parameters Action: Subscribe
	 * Topic: The topic number to subscribe LinkID: The link id of the Link which
	 * will subscribe to this topic.
	 * 
	 * Response message will have the parameters Action: Response
	 * RequestedAction: Subscribe Status: Ok or Fail Reason: a string explaining
	 * the problem if it failed.
	 */
	private void processSubscribe(Message message) {
		try {
			String linkID = message.getStringProperty(LINKID);
			if (linkID == null) {
				sendFailResponse(message, "LinkID is null.");
				return;
			}

			Link link = linkFactory.getLink(linkID);
			if (link == null) {
				sendFailResponse(message,
						"There is no RTPLink with the given linkID: " + linkID);
				return;
			}

			int topicNumber = message.getIntProperty(TOPIC);
			String error = null;

			if (link instanceof RTPLink) {
				RTPLink rtpLink = (RTPLink) link;
				error = rtpLink.subscribe(topicNumber);
			} else if (link instanceof MulticastRTPLink) {
				MulticastRTPLink mrtpLink = (MulticastRTPLink) link;
				error = mrtpLink.subscribe(topicNumber);
			}

			if (error != null) {
				sendFailResponse(message, error);
				return;
			}

			// send a success response
			sendSuccessResponse(message);

		} catch (JMSException jmse) {
			// jmse.printStackTrace();
			sendFailResponse(message, jmse.getMessage());
		}
	}

	/**
	 * processes Unsubscribe message
	 * 
	 * Received message will have the following parameters Action: Unsubscribe
	 * Topic: The topic number to unsubscribe LinkID: The link id of the Link
	 * which will unsubscribe from this topic.
	 * 
	 * Response message will have the parameters Action: Response
	 * RequestedAction: Unsubscribe Status: Ok or Fail Reason: a string
	 * explaining the problem if it fails.
	 */
	private void processUnsubscribe(Message message) {
		try {
			int topicNumber = message.getIntProperty(TOPIC);
			String linkID = message.getStringProperty(LINKID);
			if (linkID == null) {
				sendFailResponse(message, "LinkID is null.");
				return;
			}

			Link link = linkFactory.getLink(linkID);
			if (link == null) {
				sendFailResponse(message,
						"There is no RTPLink with the given linkID: " + linkID);
				return;
			}

			String error = null;

			if (link instanceof RTPLink) {
				RTPLink rtpLink = (RTPLink) link;
				error = rtpLink.unsubscribe(topicNumber);
			} else if (link instanceof MulticastRTPLink) {
				MulticastRTPLink mrtpLink = (MulticastRTPLink) link;
				error = mrtpLink.unsubscribe(topicNumber);
			}

			if (error != null) {
				sendFailResponse(message, error);
				return;
			}

			// send a success response
			sendSuccessResponse(message);

		} catch (JMSException jmse) {
			// jmse.printStackTrace();
			sendFailResponse(message, jmse.getMessage());
		}

	}

	/**
	 * send a FAIL response
	 */
	private void sendFailResponse(Message message, String reason) {
		try {
			String requestedAction = message.getStringProperty(ACTION);
			String messageID = message.getJMSMessageID();
			String responseTopic = message.getStringProperty(RESPONSE_TOPIC);

			Message responseMessage = publisherSession.createTextMessage();
			responseMessage.setStringProperty(ACTION, RESPONSE);
			responseMessage.setStringProperty(REQUESTED_ACTION, requestedAction);
			responseMessage.setStringProperty(STATUS, FAIL);
			responseMessage.setStringProperty(REASON, reason);
			responseMessage.setJMSCorrelationID(messageID);

			publishResponseMessage(responseMessage, responseTopic);

		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	/**
	 * send a Success response
	 */
	private void sendSuccessResponse(Message message) {
		sendSuccessResponse(message, null);
	}

	/**
	 * send a Success response linkID might be null
	 */
	private void sendSuccessResponse(Message message, String linkID) {
		try {
			String requestedAction = message.getStringProperty(ACTION);
			String messageID = message.getJMSMessageID();
			String responseTopic = message.getStringProperty(RESPONSE_TOPIC);

			Message responseMessage = publisherSession.createTextMessage();
			responseMessage.setStringProperty(ACTION, RESPONSE);
			responseMessage.setStringProperty(REQUESTED_ACTION, requestedAction);
			responseMessage.setStringProperty(STATUS, OK);
			if (linkID != null) {
				responseMessage.setStringProperty(LINKID, linkID);
			}
			responseMessage.setJMSCorrelationID(messageID);

			publishResponseMessage(responseMessage, responseTopic);

		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	/**
	 * publish the given message to the topic for given sessionID
	 * 
	 * @param message
	 * @param sessionID
	 */
	private void publishResponseMessage(Message message, String topic) {

		try {
			Topic tempTopic = publisherSession.createTopic(topic);
			TopicPublisher publisher = this.publisherSession
					.createPublisher(tempTopic);
			publisher.publish(message);
			publisher.close();
			logResponseMessage(message, topic);
		} catch (JMSException jmse) {
			jmse.printStackTrace();
		}
	}

	/**
	 * logs a request message sent to the given broker
	 * 
	 * @param message
	 *           the message sent
	 * @param brokerIP
	 *           the target broker ip string
	 */
	private void logResponseMessage(Message message, String tn) {

		String firstLine = "\nMessage published on: " + tn + "\n";
		String messageString = assembleMessage(message,
				RTPLinkChannelStrings.class);
		String logMessage = firstLine + messageString;
		System.out.println(logMessage);
	}

	/**
	 * logs a request message sent to the given broker
	 * 
	 * @param message
	 *           the message sent
	 * @param brokerIP
	 *           the target broker ip string
	 */
	private void logRequestMessage(Message message, String tn) {

		String firstLine = "\nMessage received on: " + tn + "\n";
		String messageString = assembleMessage(message,
				RTPLinkChannelStrings.class);
		String logMessage = firstLine + messageString;
		System.out.println(logMessage);
	}

	/**
	 * assemble the given message to be logged. This given message should be a
	 * message which has many property/value pairs. These property/value pairs
	 * are separated by ": " and each put in a line it checks all the fields in
	 * given interfaceName and assembles the set ones.
	 */
	public String assembleMessage(Message message, Class interfaceName) {

		StringBuffer buffer = new StringBuffer();
		java.lang.reflect.Field fields[] = interfaceName.getFields();

		String fieldValue = null;
		for (int i = 0; i < fields.length; i++) {
			try {
				fieldValue = (String) fields[i].get(null);
				String value = message.getStringProperty(fieldValue);
				if (value != null)
					buffer.append(fieldValue + ": " + value + "\n");
			} catch (JMSException e) {
				e.printStackTrace();
				// logger.log(Level.WARNING, "Problem when accessing message
				// property: "+fieldValue);
			} catch (Exception e) {
				e.printStackTrace();
				// logger.log(Level.WARNING, "Problem when accessing message
				// property: "+fieldValue);
			}
		}

		return buffer.toString();
	}

}
